package com.innodealing.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class BatchManualConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
    private String bootstrapServers;

//    @Value("${kafka.consumer.batch-group.group-id:batch-manual-group}")
//    private String batchGroupId;

    @Value("${kafka.consumer.batch-group.auto-offset-reset:latest}")
    private String autoOffsetReset;

    /**
     * 创建批量处理专用的消费者工厂
     */
    @Bean("batchManualConsumerFactory")
    public ConsumerFactory<String, String> batchManualConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//        props.put(ConsumerConfig.GROUP_ID_CONFIG, batchGroupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);

        // 手动提交模式 - 批量处理需要手动控制提交时机
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        
        // 批量消费相关配置
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50); // 每次拉取最多50条记录
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // 最少拉取1KB数据
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500); // 最多等待500ms
        
        // 会话和心跳配置
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);

        return new DefaultKafkaConsumerFactory<>(props);
    }

    /**
     * 配置Kafka批量监听器容器工厂，支持批量消息处理
     */
    @Bean("batchManualKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> batchManualKafkaListenerContainerFactory() {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(batchManualConsumerFactory());
        
        // 启用批量监听模式
        factory.setBatchListener(true);
        
        // 设置确认模式为手动立即确认
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        
        // 配置批量消费参数
        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setPollTimeout(1000); // 1秒超时
        
        // 配置批量错误处理器 - 替代RetryTemplate
        SeekToCurrentBatchErrorHandler errorHandler = new SeekToCurrentBatchErrorHandler();
        
        // 配置重试策略：固定间隔1秒，最多重试Integer.MAX_VALUE次（无限重试）
        FixedBackOff backOff = new FixedBackOff(1000L, Long.MAX_VALUE); // 1秒间隔，无限重试
        errorHandler.setBackOff(backOff);
        
        // 设置错误处理器
        factory.setBatchErrorHandler(errorHandler);
        
        return factory;
    }
}
